set hive.exec.dynamic.partition=true; -- # 动态分区
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.max.dynamic.partitions=3; -- 每天生成 20 个分区
set hive.exec.max.dynamic.partitions.pernode=3; --  # 每天生成 20 个分区

with tmp_all_barscan as (
    SELECT waybill_no billcode,
           scan_time scantime,
           scan_type scantype
--     FROM jms_ods.tab_barscan_taking
    from jms_dwd.dwd_tab_barscan_taking_base_dt
    WHERE dt = '{{ execution_date | cst_ds }}'
union all
         SELECT waybill_no billcode,
                scan_time scantime,
                scan_type scantype
--          FROM jms_ods.tab_barscan_warehousing
         from jms_dwd.dwd_tab_barscan_warehousing_base_dt
         WHERE dt = '{{ execution_date | cst_ds }}'
union all
         SELECT waybill_no billcode,
                scan_time scantime,
                scan_type scantype
--          FROM jms_ods.tab_barscan_bagging
         from jms_dwd.dwd_tab_barscan_bagging_base_dt
         WHERE dt = '{{ execution_date | cst_ds }}'
union all
         SELECT waybill_no billcode,
                scan_time scantime,
                scan_type scantype
--          FROM jms_ods.tab_barscan_loading
         from jms_dwd.dwd_tab_barscan_loading_base_dt
         WHERE dt = '{{ execution_date | cst_ds }}'
union all
         SELECT waybill_no billcode,
                scan_time scantime,
                scan_type scantype
--          FROM jms_ods.tab_barscan_send
         from jms_dwd.dwd_tab_barscan_send_base_dt
         WHERE dt = '{{ execution_date | cst_ds }}'
union all
         SELECT waybill_no billcode,
                scan_time scantime,
                scan_type scantype
--          FROM jms_ods.tab_barscan_arrival
         from jms_dwd.dwd_tab_barscan_arrival_base_dt
         WHERE dt = '{{ execution_date | cst_ds }}'
union all
         SELECT waybill_no billcode,
                scan_time scantime,
                scan_type scantype
--          FROM jms_ods.tab_barscan_unloading
         from jms_dwd.dwd_tab_barscan_unloading_base_dt
         WHERE dt = '{{ execution_date | cst_ds }}'
union all
         SELECT waybill_no billcode,
                scan_time scantime,
                scan_type scantype
--          FROM jms_ods.tab_barscan_difficult
         from jms_dwd.dwd_tab_barscan_difficult_base_dt
         WHERE dt = '{{ execution_date | cst_ds }}'
union all
         SELECT waybill_no billcode,
                scan_time scantime,
                scan_type scantype
--          FROM jms_ods.tab_barscan_deliver
         from jms_dwd.dwd_tab_barscan_deliver_base_dt
         WHERE dt = '{{ execution_date | cst_ds }}'
union all
         SELECT waybill_no billcode,
                scan_time scantime,
                scan_type scantype
--          FROM jms_ods.tab_barscan_collect
         from jms_dwd.dwd_tab_barscan_collect_base_dt
         WHERE dt = '{{ execution_date | cst_ds }}'
union all
         select waybill_no billcode,
                signed_time as scantime,
                "派件扫描"      as scantype
--          from jms_ods.tab_barscan_sign
         from jms_dwd.dwd_tab_barscan_sign_base_dt
         where dt = '{{ execution_date | cst_ds }}'
union all
         select waybill_no as billcode,
                print_time as scantime,
                "退件扫描"   as scantype
--          from jms_ods.tab_reback_transfer_express
         from jms_dwd.dwd_tab_reback_transfer_express_base
--          where dt = '{{ execution_date | cst_ds }}'
union all
         SELECT waybill_no billcode,
                scan_time scantime,
                scan_type scantype
--          FROM jms_ods.tab_barscan_other
         from jms_dwd.dwd_tab_barscan_other_base_dt
         WHERE dt = '{{ execution_date | cst_ds }}'
union all
         SELECT waybill_no,
                cast(scantime AS TIMESTAMP) AS scantime,
                scan_type_name              as scantype
         -- FROM jms_ods.yl_oms_interceptorpiece LATERAL VIEW explode(split(
          FROM jms_dwd.dwd_yl_oms_interceptorpiece_base LATERAL VIEW explode(split(
                 concat_ws(',', cast(registration_time AS string), cast(intercept_time AS string),
                           cast(cancal_time AS string)), ",")) lve AS scantime
         -- WHERE dt = '2023-08-13'
     )
INSERT overwrite TABLE jms_dws.dws_end_piece_trace_track_dt partition(dt)
SELECT tmp_1.waybill_no,
       tmp_1.order_id,
       tmp_1.end_code,
       tmp_1.end_code_desc,
       tmp_1.source,
       tmp_1.end_time,
       tmp_1.scantime,
       tmp_1.scantype,
       tmp_1.flag,
       '{{ execution_date | cst_ds }}' AS dt
FROM
  ( SELECT tmp_0.*,
           row_number() over w1 rank
   FROM
     (SELECT tmp_end_piece.waybill_no,
             tmp_end_piece.end_code,
             tmp_end_piece.end_code_desc,
             tmp_end_piece.source,
             tmp_end_piece.end_time,
             null as order_id,
             tmp_all_barscan.scantime,
             tmp_all_barscan.scantype,
             if(tmp_all_barscan.billcode IS NULL,0, 1) AS flag
      FROM
        (SELECT waybill_no,
                end_code,
                end_code_desc,
                SOURCE,
                end_time
         FROM jms_dwd.dwd_tab_end_piece_unprocessed_dt
         WHERE dt='{{ execution_date | cst_ds }}'
         UNION ALL
         SELECT waybill_no,
                          end_code,
                          end_code_desc,
                          SOURCE,
                          end_time
         FROM jms_dws.dws_end_piece_trace_track_dt
         WHERE dt='{{ execution_date | date_add(-1) | cst_ds }}'
           AND flag = 0 )   tmp_end_piece
      LEFT JOIN tmp_all_barscan
         ON tmp_end_piece.waybill_no = tmp_all_barscan.billcode AND tmp_all_barscan.scantime > tmp_end_piece.end_time)
         tmp_0 window w1 AS (partition BY waybill_no ORDER BY scantime ASC)) tmp_1
         WHERE rank = 1  and flag = 1
         distribute by dt;